package com.rrd.dw.mr.text;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.rrd.dw.utils.HadoopUtils;
import com.rrd.dw.utils.MD5Util;

/**
 * @author liubaoxin
 * @version 创建时间：2017年11月10日 上午10:44:43
 *          解析同盾明细数据，将JSON数据转换为各个维度组合输出多行
 */
public class TongdunDetailMR extends Configured implements Tool {
    public static final String CRT_FLAG = "\001";
    public static final Pattern CRT_PATTERN = Pattern.compile(CRT_FLAG);

    public static class TongdunMapper
            extends Mapper<LongWritable, Text, NullWritable, Text> {

        private Map<String, String> pdlMapAndroid = null;
        private Map<String, String> pdlMapIos = null;
        private Map<String, String> rrdMapAndroid = null;
        private Map<String, String> rrdMapIos = null;
        private Text text = new Text();
        private String valStr = "";

        private void initMap() {
            pdlMapAndroid = new HashMap<String, String>();
            pdlMapIos = new HashMap<String, String>();
            rrdMapAndroid = new HashMap<String, String>();
            rrdMapIos = new HashMap<String, String>();
            pdlMapAndroid.put("2794841", "7天内");
            pdlMapAndroid.put("2794843", "1个月内");
            pdlMapAndroid.put("2794845", "3个月内");
            pdlMapAndroid.put("2794847", "6个月内");
            pdlMapAndroid.put("2794849", "12个月内");
            pdlMapAndroid.put("2794851", "18个月内");
            pdlMapAndroid.put("2794853", "24个月内");
            pdlMapAndroid.put("2794855", "近60个月以上");
            //pdl ios
            pdlMapIos.put("2794437", "7天内");
            pdlMapIos.put("2794439", "1个月内");
            pdlMapIos.put("2794441", "3个月内");
            pdlMapIos.put("2794443", "6个月内");
            pdlMapIos.put("2794445", "12个月内");
            pdlMapIos.put("2794447", "18个月内");
            pdlMapIos.put("2794449", "24个月内");
            pdlMapIos.put("2794451", "近60个月以上");
            //rrd android
            rrdMapAndroid.put("3063537", "7天内");
            rrdMapAndroid.put("3063539", "1个月内");
            rrdMapAndroid.put("3063541", "3个月内");
            //rrd ios
            rrdMapIos.put("3063379", "7天内");
            rrdMapIos.put("3063381", "1个月内");
            rrdMapIos.put("3063383", "3个月内");
        }

        @Override
        protected void setup(
                Mapper<LongWritable, Text, NullWritable, Text>.Context context)
                throws IOException, InterruptedException {
            initMap();
        }

        private String[] tmparr = null;
        private List<String> list = new ArrayList<String>(200);
        private String period = null;

        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            if (value == null) {
                return;
            }
            tmparr = CRT_PATTERN.split(value.toString());
            if (tmparr.length < 8) {
                return;
            }
            //判断是否是同盾detail 
            if (tmparr[3] == null || !tmparr[3].equals("TONGDUN_DETAILS")
                || tmparr[1] == null) {
                return;
            }
            //判断业务线
            if (!tmparr[1].equals("PAY_DAY_LOAN")
                && !tmparr[1].equals("REN_REN_DAI")
                && !tmparr[1].equals("HAO_HUAN")) {
                return;
            }
            list.clear();
            parseJson(tmparr[1], tmparr[4]);
            //将同盾相关信息写入文件
            valStr = tmparr[0] + CRT_FLAG + tmparr[1] + CRT_FLAG + tmparr[2]
                + CRT_FLAG;
            for (String s : list) {
                text.set(MD5Util.getMd5(valStr + s + tmparr[5]) + CRT_FLAG
                    + valStr + s + CRT_FLAG + tmparr[5] + CRT_FLAG + tmparr[6]);
                context.write(NullWritable.get(), text);

            }
        }

        public void parseJson(String chnl, String str) {
            JSONObject obj = JSON.parseObject(str);
            JSONArray rules = obj.getJSONArray("rules");
            JSONObject tmpObj = null;
            String ruleId = "";
            if (rules == null) {
                return;
            }
            for (int i = 0; i < rules.size(); i++) {
                tmpObj = rules.getJSONObject(i);
                ruleId = tmpObj.getString("ruleId");

                if (chnl.equals("REN_REN_DAI") && rrdMapIos.get(ruleId) == null
                    && rrdMapAndroid.get(ruleId) == null) {
                    //开始解析json内容
                    continue;
                }
                if ((chnl.equals("PAY_DAY_LOAN") || chnl.equals("HAO_HUAN"))
                    && pdlMapIos.get(ruleId) == null
                    && pdlMapAndroid.get(ruleId) == null) {
                    //开始解析json内容
                    continue;
                }

                if (chnl.equals("REN_REN_DAI")) {
                    period = rrdMapIos.get(ruleId) == null
                        ? rrdMapAndroid.get(ruleId)
                        : rrdMapIos.get(ruleId);
                }
                if (chnl.equals("PAY_DAY_LOAN") || chnl.equals("HAO_HUAN")) {
                    period = pdlMapIos.get(ruleId) == null
                        ? pdlMapAndroid.get(ruleId)
                        : pdlMapIos.get(ruleId);
                }
                //解析hits数组
                JSONArray conditions = tmpObj.getJSONArray("conditions");
                JSONObject conditionsObj = null;
                if (conditions == null) {
                    return;
                }
                for (int j = 0; j < conditions.size(); j++) {
                    conditionsObj = conditions.getJSONObject(j);
                    list.add(
                        "ALL" + CRT_FLAG + "ALL" + CRT_FLAG + period + CRT_FLAG
                            + (conditionsObj.getString("result") == null ? 0
                                : conditionsObj.getString("result")));
                    parseJson("hits", conditionsObj.getJSONArray("hits"));
                    parseJson("hitsForDim",
                        conditionsObj.getJSONArray("hitsForDim"));
                    parseJson("resultsForDim",
                        conditionsObj.getJSONArray("resultsForDim"));
                }
            }
        }

        private void parseJson(String tagName, JSONArray jsonArray) {
            StringBuilder sb = new StringBuilder(100);
            JSONObject obj = null;
            String accountType = null;

            for (int i = 0; i < jsonArray.size(); i++) {
                obj = jsonArray.getJSONObject(i);

                sb.delete(0, sb.length());
                sb.append(obj.getString("industryDisplayName") == null ? "ALL"
                    : obj.getString("industryDisplayName")).append(CRT_FLAG);
                accountType = obj.getString("dimType");
                if (accountType == null || accountType.equals("")) {
                    accountType = obj.getString("matchDimType") == null
                        ? obj.getString("originalDimType")
                        : obj.getString("matchDimType");

                }
                sb.append(accountType == null ? "ALL" : accountType)
                    .append(CRT_FLAG);
                sb.append(period).append(CRT_FLAG);

                sb.append(obj.getString("count") == null ? "ALL"
                    : obj.getString("count"));

                list.add(sb.toString());
            }
        }
    }

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        Job job = Job.getInstance(conf);

        String inputDir = conf.get("param.input.path",
            "/data/bdm/pdl03_third_party_data_inc/dt=2017-09-25");
        String outDir = conf.get("param.out.path",
            "/tmp/data/fdm/tongdun_detail_inc");
        System.out.println("param.input.path=" + inputDir);
        System.out.println("param.out.path=" + outDir);
        Path outpath = new Path(outDir);
        HadoopUtils.delete(conf, outpath);
        TextInputFormat.setInputPaths(job, new Path(inputDir));
        TextOutputFormat.setOutputPath(job, new Path(outDir));
        job.setOutputKeyClass(NullWritable.class);

        job.setJarByClass(TongdunDetailMR.class);
        job.setMapperClass(TongdunMapper.class);

        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new TongdunDetailMR(), args);
        System.exit(res);
    }

}
